/**
 * 
 */
package nio2;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

/**
 * @author wudan_js
 *
 */
public class NioThreadServer {
	private static ByteBuffer buffer=ByteBuffer.allocate(1024);//1K大小的缓冲池  
	private static final int MAX_THREADS = 5; 
	private static ThreadPool pool = new ThreadPool(MAX_THREADS);

	public static void main(String[] args) {
		try {  
            Selector selector=Selector.open();  
            ServerSocketChannel serverChannel=ServerSocketChannel.open();  
            ServerSocket server=serverChannel.socket();  
            server.bind(new InetSocketAddress(12345));  
            serverChannel.configureBlocking(false);  
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);  
              
            while(true){  
                int n=selector.select();  
                if (n == 0) {    // nothing to do  
                    continue;  
                }  
                Iterator<SelectionKey> iterator=selector.selectedKeys().iterator();  
                while(iterator.hasNext()){  
                    SelectionKey key=iterator.next();  
                    // Is a new connection coming in?  
                    if(key.isAcceptable()){  
                        //此时的key是，ServerSocketChannel注册的。要获取客户端通道  
                        ServerSocketChannel serverSocketChannel=(ServerSocketChannel)key.channel();  
                        SocketChannel client=serverSocketChannel.accept();  
                        client.configureBlocking(false);//将客户端通道也设置成非阻塞  
                        client.register(selector, SelectionKey.OP_READ);  
                        //向客户端发送消息  
                        buffer.clear();
                        buffer.put("Hi there!\r\n".getBytes());
                        buffer.flip();
                        client.write(buffer);
                    }  
                    if(key.isReadable()){  
                    	WorkerThread worker = pool.getWorker(); 
                		if (worker == null) { 
                			//没有可用线程。什么也不做。
                			//选择循环将不断调用这个方法，直到线程变得可用。这样的设计可以得到改善。
                			return; 
                		}
                		// 调用此唤醒工作线程，然后返回
                		worker.serviceChannel(key); 
                    }  
                    iterator.remove();  
                }  
            }  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
	}
	
	/**
	 *  一个非常简单的线程池类。池大小设置为 MAX_THREADS
	 *  ，并保持固定。主题是通过FIFO空闲队列循环. 
	 */
	private static class ThreadPool { 
		List<WorkerThread> idle = new LinkedList<WorkerThread>(); 
		ThreadPool(int poolSize) { 
			// Fill up the pool with worker threads 
			for (int i = 0; i < poolSize; i++) { 
				WorkerThread thread = new WorkerThread(this); 
				// Set thread name for debugging. Start it. 
				thread.setName("Worker" + (i + 1)); 
				thread.start(); 
				idle.add(thread); 
			}
		}	
		/** 找到一个空闲的工作线程（如有）。可能返回null。 */ 
		WorkerThread getWorker() { 
			WorkerThread worker = null; 
			synchronized (idle) { 
				if (idle.size() > 0) {
					worker = (WorkerThread) idle.remove(0); 
				} 
			}
			return (worker);
		}
		/** 由工人线程调用，返回其空闲池。 */ 
		void returnWorker(WorkerThread worker) { 
			synchronized (idle) { 
				idle.add(worker); 
			} 
		}
	}
	
	private static class WorkerThread extends Thread { 
		private ByteBuffer buffer = ByteBuffer.allocate(1024); 
		private ThreadPool pool; 
		private SelectionKey key; 
		WorkerThread(ThreadPool pool) { 
			this.pool = pool; 
		}
		// Loop forever waiting for work to do 
		public synchronized void run() {
			System.out.println(this.getName() + " is ready"); 
			while (true) {
				try { 
					// Sleep and release object lock 
					this.wait(); 
				} catch (InterruptedException e) { 
					e.printStackTrace(); 
					// Clear interrupt status 
					Thread.interrupted(); 
				}
				if (key == null) { 
					continue; 
					// just in case 
				} 
				System.out.println(this.getName() + " has been awakened"); 
				try { 
					drainChannel(key); 
				} catch (Exception e) { 
					System.out.println("Caught '" + e + "' closing channel"); 
					// Close channel and nudge selector 
					try { 
						key.channel().close(); 
					} catch (IOException ex) { 
						ex.printStackTrace(); 
					}key.selector().wakeup(); 
				}
				key = null; 
				// Done. Ready for more. Return to pool 
				this.pool.returnWorker(this); 
			}
		}
		
		synchronized void serviceChannel(SelectionKey key) { 
			this.key = key; 
			key.interestOps(key.interestOps() & (~SelectionKey.OP_READ)); 
			this.notify(); 
			// Awaken the thread 
		}
		
		void drainChannel(SelectionKey key) throws Exception { 
			SocketChannel channel = (SocketChannel) key.channel(); 
			int count; 
			buffer.clear(); 
			// Empty buffer 
			// Loop while data is available; channel is nonblocking 
			while ((count = channel.read(buffer)) > 0) { 
				buffer.flip(); 
				// make buffer readable 
				// Send the data; may not go all at once 
				while (buffer.hasRemaining()) { 
					channel.write(buffer); 
				}
				// WARNING: the above loop is evil. 
				// See comments in superclass. 
				buffer.clear(); 
				// Empty buffer 
			}
			if (count < 0) { 
				// Close channel on EOF; invalidates the key 
				channel.close(); 
				return; 
			}
			// Resume interest in OP_READ 
			key.interestOps(key.interestOps() | SelectionKey.OP_READ); 
			// Cycle the selector so this key is active again 
			key.selector().wakeup(); //
		}
	}
}
